// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: ziyan
#include "coin2/exchange/bittrex/feed_v3/processor.h"

#include <string>
#include <utility>

#include "coin2/base/base64.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/base/api_util/zlib_decompress.h"
#include "coin2/exchange/bittrex/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

using impl::base64_decode;
namespace coin2::exchange::bittrex::feed_v3::impl {

void BittrexFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);
  rapidjson::Document wrapped_document;

  wrapped_document.Parse(packet.payload.data());

  if (wrapped_document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(wrapped_document.GetParseError())
            << packet.payload;
  }

  if (wrapped_document.ObjectEmpty()) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(wrapped_document);
  }

  std::string channel;
  std::vector<std::string> url_para;
  std::string native_symbol;
  rapidjson::Document document;

  if (wrapped_document.HasMember("M") && wrapped_document["M"].Size() > 0) {
    channel = wrapped_document["M"][0]["M"].GetString();
    if (channel == "orderBook") {channel = "book";}
    std::string base64_str = wrapped_document["M"][0]["A"][0].GetString();
    auto json_str = deflate_string(base64_decode(base64_str));
    document.Parse(json_str.c_str());
    if (document.HasParseError()) {
      THROW() << "JSON parse error: "
              << GetParseError_En(document.GetParseError())
              << json_str;
    }
    native_symbol = document["marketSymbol"].GetString();
  } else if (wrapped_document.HasMember("url")) {
    url_para = Split(wrapped_document["url"].GetString(), "/");
    if (url_para.size() > 6 && url_para[6] == "candles") {
      channel = "kline_rest";
      native_symbol = url_para[5];
    }
  }

  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!upd_mgr_.is_symbol_registered(native_symbol)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (channel == "book") {
    auto book = upd_mgr_.GetBookMutable(native_symbol);
    if (book != nullptr) {
      BittrexFeedUpdateParser::ParseBook(book, document);
    }
    upd_mgr_.PublishBook();
  } else if (channel == "trade") {
    Trade* agg_trade = upd_mgr_.GetTradeMutable();
    agg_trade->Clear();
    Trade trade{};
    for (const auto& trade_json : document["deltas"].GetArray()) {
      BittrexFeedUpdateParser::ParseTrade(&trade, trade_json);
      if (TryAggregateTrades(agg_trade, trade)) {
        continue;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, true);
      }
      *agg_trade = trade;
    }
    if (agg_trade->fill_qty > 0) {
      upd_mgr_.PublishTrade(native_symbol, false);
    }
  } else if (channel == "kline_rest") {
    auto interval = BittrexFeedUpdateParser::ParseKlineInterval(url_para[7]);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      bool to_publish = false;
      if (klines->Size() == 0) {
        // first query
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      for (const auto& msg : wrapped_document["data"].GetArray()) {
        bool freezed = BittrexFeedUpdateParser::ParseKlineRest(msg, klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    LOG(INFO) << "[Bittrex Feed Processor] unknown msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::bittrex::feed_v3::impl